package com.hkdw.wc.apitest.window;/**
 * Copyright (c) 2018-2028 尚硅谷 All Rights Reserved
 * <p>
 * Project: FlinkTutorial
 * Package: com.atguigu.apitest.window
 * Version: 1.0
 * <p>
 * Created by wushengran on 2020/11/9 14:37
 */

import com.hkdw.wc.apitest.beans.SensorReading;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @ClassName: WindowTest1_TimeWindow
 * @Description:
 * @Author: wushengran on 2020/11/9 14:37
 * @Version: 1.0
 */
public class WindowTest1_TimeWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        // 从文件读取数据
//        DataStream<String> inputStream = env.readTextFile("D:\\Projects\\BigData\\FlinkTutorial\\src\\main\\resources\\sensor.txt");

        // socket文本流
        DataStream<String> inputStream = env.socketTextStream("localhost", 7777);

        // 转换成SensorReading类型
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        // 开窗测试

        // 1. 增量聚合函数
        //dataStream.windowAll()  : 是把所有的数据都放到同一个窗口
        DataStream<Integer> resultStream = dataStream.keyBy("id") //keyBy 做了一个分组
//                .countWindow(10, 2);
//                .window(EventTimeSessionWindows.withGap(Time.minutes(1)));
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(15)))
                .timeWindow(Time.seconds(15))  //一个参数代表滚动窗口 两个参数代表滑动窗口 ，session window就要使用window了
//                .countWindow(10,2)  ：10 代表窗口大小，2是滑动的步长 （）
//                .sum(1) //都是增量聚合
                //AggregateFunction<输入，累加器（得到的结果），输出>
                .aggregate(new AggregateFunction<SensorReading, Integer, Integer>() {
                    //创建一个累加器 给0
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }

                    //应该怎么累加
                    @Override
                    public Integer add(SensorReading value, Integer accumulator) {
                        return accumulator + 1;
                    }

                    //输出结果
                    @Override
                    public Integer getResult(Integer accumulator) {
                        return accumulator;
                    }

                    //一般时间窗口用不到，一般是session窗口使用，用作与合并
                    @Override
                    public Integer merge(Integer a, Integer b) {
                        return a + b;
                    }
                });


        /***
         *
         * • .trigger() —— 触发器
         * ➢ 定义 window 什么时候关闭，触发计算并输出结果
         * • .evictor() —— 移除器
         * ➢ 定义移除某些数据的逻辑
         * • .allowedLateness() —— 允许处理迟到的数据
         * • .sideOutputLateData() —— 将迟到的数据放入侧输出流
         * • .getSideOutput() —— 获取侧输出流
         */
        // 2. 全窗口函数
        SingleOutputStreamOperator<Tuple3<String, Long, Integer>> resultStream2 = dataStream.keyBy("id")
                .timeWindow(Time.seconds(15))
//                .process(new ProcessWindowFunction<SensorReading, Object, Tuple, TimeWindow>() {
//                })
                //WindowFunction<输入，输出，分组的key,window的信息>
                .apply(new WindowFunction<SensorReading, Tuple3<String, Long, Integer>, Tuple, TimeWindow>() {
                    @Override
                    //apply<当前的key, window信息, 当前输入值，当前输出值>
                    public void apply(Tuple tuple, TimeWindow window, Iterable<SensorReading> input, Collector<Tuple3<String, Long, Integer>> out) throws Exception {
                        String id = tuple.getField(0);
                        Long windowEnd = window.getEnd();
                        Integer count = IteratorUtils.toList(input.iterator()).size();
                        out.collect(new Tuple3<>(id, windowEnd, count));
                    }
                });

        // 3. 其它可选API
        OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("late") {
        };

        SingleOutputStreamOperator<SensorReading> sumStream = dataStream.keyBy("id")
                .timeWindow(Time.seconds(15))
//                .trigger()
//                .evictor()
                .allowedLateness(Time.minutes(1))//延迟时间，允许时间到了先计算结果，但是不关闭窗口，在等1分钟，来一个就计算一次（叠加一次）
                .sideOutputLateData(outputTag) // 如果上面延迟窗口也关闭了，可以把数据延迟放到侧输出流里面
                .sum("temperature");

        sumStream.getSideOutput(outputTag).print("late");

        resultStream.print();


        env.execute();
    }
}
